package com.ksyun.campus.metaserver.monitoring;

import com.ksyun.campus.metaserver.domain.BlockInfo;
import com.ksyun.campus.metaserver.domain.DataServerInfo;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * 副本放置策略监控和指标收集
 * 用于跟踪和分析三副本分配的效果
 */
@Slf4j
@Component
public class ReplicaPlacementMetrics {
    
    // 分配统计
    private final AtomicLong totalAllocations = new AtomicLong(0);
    private final AtomicLong successfulAllocations = new AtomicLong(0);
    private final AtomicLong failedAllocations = new AtomicLong(0);
    
    // 负载均衡统计
    private final Map<String, AtomicLong> dataServerAllocations = new ConcurrentHashMap<>();
    private final Map<String, AtomicLong> rackAllocations = new ConcurrentHashMap<>();
    
    // 性能统计
    private final List<AllocationEvent> recentAllocations = Collections.synchronizedList(new ArrayList<>());
    private final int MAX_RECENT_EVENTS = 1000;
    
    /**
     * 记录块分配事件
     */
    public void recordAllocation(String fileId, int blkIdx, List<BlockInfo.ReplicaInfo> replicas, 
                               long allocationTimeMs, boolean success) {
        totalAllocations.incrementAndGet();
        
        if (success) {
            successfulAllocations.incrementAndGet();
            
            // 统计DataServer分配次数
            for (BlockInfo.ReplicaInfo replica : replicas) {
                dataServerAllocations.computeIfAbsent(replica.getDsId(), k -> new AtomicLong(0))
                        .incrementAndGet();
            }
            
            // 创建分配事件记录
            AllocationEvent event = new AllocationEvent();
            event.fileId = fileId;
            event.blkIdx = blkIdx;
            event.replicaCount = replicas.size();
            event.allocationTimeMs = allocationTimeMs;
            event.timestamp = LocalDateTime.now();
            event.dataServerIds = replicas.stream()
                    .map(BlockInfo.ReplicaInfo::getDsId)
                    .collect(Collectors.toList());
            
            // 添加到最近事件列表
            synchronized (recentAllocations) {
                recentAllocations.add(event);
                if (recentAllocations.size() > MAX_RECENT_EVENTS) {
                    recentAllocations.remove(0);
                }
            }
            
        } else {
            failedAllocations.incrementAndGet();
        }
        
        // 定期输出统计信息
        if (totalAllocations.get() % 100 == 0) {
            logStatistics();
        }
    }
    
    /**
     * 记录机架分配统计
     */
    public void recordRackAllocation(List<DataServerInfo> selectedServers) {
        for (DataServerInfo server : selectedServers) {
            String rack = getRackId(server);
            if (rack != null) {
                rackAllocations.computeIfAbsent(rack, k -> new AtomicLong(0))
                        .incrementAndGet();
            }
        }
    }
    
    /**
     * 获取负载均衡统计信息
     */
    public LoadBalanceStats getLoadBalanceStats() {
        LoadBalanceStats stats = new LoadBalanceStats();
        
        // 计算DataServer分配均匀度
        if (!dataServerAllocations.isEmpty()) {
            Collection<AtomicLong> allocations = dataServerAllocations.values();
            long totalAllocs = allocations.stream().mapToLong(AtomicLong::get).sum();
            double avgAllocations = (double) totalAllocs / allocations.size();
            
            // 计算标准差
            double variance = allocations.stream()
                    .mapToDouble(count -> Math.pow(count.get() - avgAllocations, 2))
                    .average()
                    .orElse(0.0);
            stats.dataServerStdDev = Math.sqrt(variance);
            stats.dataServerAvgAllocations = avgAllocations;
            
            // 计算最大最小分配比例
            long maxAllocs = allocations.stream().mapToLong(AtomicLong::get).max().orElse(0);
            long minAllocs = allocations.stream().mapToLong(AtomicLong::get).min().orElse(0);
            stats.maxMinRatio = minAllocs > 0 ? (double) maxAllocs / minAllocs : Double.MAX_VALUE;
        }
        
        // 计算机架分配均匀度
        if (!rackAllocations.isEmpty()) {
            Collection<AtomicLong> rackAllocs = rackAllocations.values();
            long totalRackAllocs = rackAllocs.stream().mapToLong(AtomicLong::get).sum();
            double avgRackAllocations = (double) totalRackAllocs / rackAllocs.size();
            
            double rackVariance = rackAllocs.stream()
                    .mapToDouble(count -> Math.pow(count.get() - avgRackAllocations, 2))
                    .average()
                    .orElse(0.0);
            stats.rackStdDev = Math.sqrt(rackVariance);
            stats.rackAvgAllocations = avgRackAllocations;
        }
        
        return stats;
    }
    
    /**
     * 获取性能统计信息
     */
    public PerformanceStats getPerformanceStats() {
        PerformanceStats stats = new PerformanceStats();
        
        synchronized (recentAllocations) {
            if (!recentAllocations.isEmpty()) {
                // 计算分配时间统计
                List<Long> allocationTimes = recentAllocations.stream()
                        .map(event -> event.allocationTimeMs)
                        .sorted()
                        .collect(Collectors.toList());
                
                stats.avgAllocationTimeMs = allocationTimes.stream()
                        .mapToLong(Long::longValue)
                        .average()
                        .orElse(0.0);
                
                int size = allocationTimes.size();
                stats.p50AllocationTimeMs = allocationTimes.get(size / 2);
                stats.p95AllocationTimeMs = allocationTimes.get((int) (size * 0.95));
                stats.p99AllocationTimeMs = allocationTimes.get((int) (size * 0.99));
                
                // 计算成功率
                stats.successRate = (double) successfulAllocations.get() / totalAllocations.get();
                
                // 计算吞吐量（最近1分钟）
                LocalDateTime oneMinuteAgo = LocalDateTime.now().minusMinutes(1);
                long recentCount = recentAllocations.stream()
                        .filter(event -> event.timestamp.isAfter(oneMinuteAgo))
                        .count();
                stats.allocationsPerMinute = recentCount;
            }
        }
        
        return stats;
    }
    
    /**
     * 获取详细的分配分布信息
     */
    public Map<String, Object> getDetailedStats() {
        Map<String, Object> stats = new HashMap<>();
        
        // 基础统计
        stats.put("totalAllocations", totalAllocations.get());
        stats.put("successfulAllocations", successfulAllocations.get());
        stats.put("failedAllocations", failedAllocations.get());
        stats.put("successRate", (double) successfulAllocations.get() / Math.max(1, totalAllocations.get()));
        
        // DataServer分配分布
        Map<String, Long> dsDistribution = dataServerAllocations.entrySet().stream()
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        entry -> entry.getValue().get()
                ));
        stats.put("dataServerDistribution", dsDistribution);
        
        // 机架分配分布
        Map<String, Long> rackDistribution = rackAllocations.entrySet().stream()
                .collect(Collectors.toMap(
                        Map.Entry::getKey,
                        entry -> entry.getValue().get()
                ));
        stats.put("rackDistribution", rackDistribution);
        
        // 负载均衡和性能统计
        stats.put("loadBalanceStats", getLoadBalanceStats());
        stats.put("performanceStats", getPerformanceStats());
        
        return stats;
    }
    
    /**
     * 重置统计信息
     */
    public void reset() {
        totalAllocations.set(0);
        successfulAllocations.set(0);
        failedAllocations.set(0);
        dataServerAllocations.clear();
        rackAllocations.clear();
        
        synchronized (recentAllocations) {
            recentAllocations.clear();
        }
        
        log.info("Replica placement metrics reset");
    }
    
    /**
     * 输出统计日志
     */
    private void logStatistics() {
        LoadBalanceStats lbStats = getLoadBalanceStats();
        PerformanceStats perfStats = getPerformanceStats();
        
        log.info("Replica Placement Statistics - Total: {}, Success: {}, Failed: {}, Success Rate: {:.2f}%",
                totalAllocations.get(),
                successfulAllocations.get(),
                failedAllocations.get(),
                perfStats.successRate * 100);
        
        log.info("Load Balance - DataServer StdDev: {:.2f}, Rack StdDev: {:.2f}, Max/Min Ratio: {:.2f}",
                lbStats.dataServerStdDev,
                lbStats.rackStdDev,
                lbStats.maxMinRatio);
        
        log.info("Performance - Avg Time: {:.2f}ms, P95: {}ms, Throughput: {} allocs/min",
                perfStats.avgAllocationTimeMs,
                perfStats.p95AllocationTimeMs,
                perfStats.allocationsPerMinute);
    }
    
    private String getRackId(DataServerInfo ds) {
        if (ds.getLabels() != null) {
            return ds.getLabels().get("rack");
        }
        return null;
    }
    
    // 内部数据结构
    @Data
    public static class LoadBalanceStats {
        private double dataServerStdDev;
        private double dataServerAvgAllocations;
        private double rackStdDev;
        private double rackAvgAllocations;
        private double maxMinRatio;
    }
    
    @Data
    public static class PerformanceStats {
        private double avgAllocationTimeMs;
        private long p50AllocationTimeMs;
        private long p95AllocationTimeMs;
        private long p99AllocationTimeMs;
        private double successRate;
        private long allocationsPerMinute;
    }
    
    private static class AllocationEvent {
        String fileId;
        int blkIdx;
        int replicaCount;
        long allocationTimeMs;
        LocalDateTime timestamp;
        List<String> dataServerIds;
    }
}
